package org.example

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

object lianxi3 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local[*]").appName("spark").getOrCreate()
    val sc = spark.sparkContext

    // 表结构读取电影用户数据
    val schemaUser = StructType(Seq(
      StructField("id", IntegerType, nullable = false),
      StructField("gender", StringType),
      StructField("age", StringType),
      StructField("occupation", IntegerType),
      StructField("location", StringType)
    ))
    val user = spark.read.option("sep", "::").schema(schemaUser)
      .csv("src/main/resources/users.dat")

    // 读取评分数据
    val schemaRating = StructType(Seq(
      StructField("userId", IntegerType, nullable = false),
      StructField("movieId", IntegerType, nullable = false),
      StructField("rating", IntegerType),
      StructField("timestamp", StringType)
    ))
    val rating = spark.read.option("sep", "::").schema(schemaRating)
      .csv("src/main/resources/ratings.dat")

    // 读取电影数据
    val schemaMovie = StructType(Seq(
      StructField("movieId", IntegerType, nullable = false),
      StructField("title", StringType),
      StructField("genres", StringType)
    ))
    val movie = spark.read.option("sep", "::").schema(schemaMovie)
      .csv("src/main/resources/movies.dat")

    // 查询18岁女生评分电影为5分的所有电影名
    val result = user
      .where("gender = 'F' and age = '18'")
      .join(rating, user("id") === rating("userId"))
      .where("rating = 5")
      .join(movie, rating("movieId") === movie("movieId"))
      .select("title")
      .distinct()

    result.show()

    sc.stop()
  }
}
